#include "impl_redis.h"

#include "common/scope_guard.h"
#include "messaging/exception.h"


#pragma GCC diagnostic ignored "-Wunused-parameter"
#pragma GCC diagnostic ignored "-Wunused-variable"

namespace evm { namespace messaging {

RedisAsync::RedisAsync(const string& name, const string &ip, int port,
                       EventLoopPtr loop, size_t retry_time)
    : name(name), ip(ip), port(port), loop(loop), retry_time(retry_time), ctx(nullptr)
{
    read_handler = [this](int, int) {
        lock_guard<recursive_mutex> lg(rw_mutex);
        redisAsyncHandleRead(ctx);
    };

    write_handler = [this](int, int) {
        lock_guard<recursive_mutex> lg(rw_mutex);
        redisAsyncHandleWrite(ctx);
    };
}

void RedisAsync::start(bool wait) {
    log_info("{}, start", name);
    unique_lock<recursive_mutex> lk(conn_mutex);

    while (state != CONNECTED) {
        connect();

        if (!wait || state == CONNECTED)
            break;

        conn_cv.wait(lk);
    }

    log_info("{}, started", name);
}

void RedisAsync::stop(bool wait) {
    log_info("{}, stop", name);
    unique_lock<recursive_mutex> lk(conn_mutex);

    while (state == CONNECTED) {
        disconnect();

        if (!wait || state != CONNECTED)
            break;

        conn_cv.wait_for(lk, chrono::milliseconds(POLL_TIMEOUT));
    }

    log_info("{}, stopped", name);
}

void RedisAsync::connect() {
    log_info("{}, connect: {}:{}", name, ip, port);

    /* New async context. */
    ctx = redisAsyncConnect(ip.data(), port);
    if (!ctx || ctx->err)
        throw_with_log("redis async connect error: {}",
                       !ctx ? "null" : static_cast<const char*>(ctx->errstr));

    /* Store the socket file descriptor. */
    fd = ctx->c.fd;

    /* Register functions to start/stop listening for events.
     * `this` is passed as the (void*)data. */
    ctx->ev = {this, add_read, del_read, add_write, del_write, cleanup};

    /* Set the connect and disconnect callback. */
    redisAsyncSetConnectCallback(ctx, connected);  // this will also call add_write to add fd to event loop
    redisAsyncSetDisconnectCallback(ctx, disconnected);
}

void RedisAsync::disconnect() {
    /* Initiate disconnection. */
    redisAsyncDisconnect(ctx);
}

void RedisAsync::reconnect() {
    loop->add_timer(retry_time, [this](size_t id) {
        connect();
        return EV_NOMORE;  // one shot timer
    });
}

void RedisAsync::connected(const redisAsyncContext *ctx, int status) {
    auto r = static_cast<RedisAsync*>(ctx->ev.data);

    lock_guard<recursive_mutex> lg(r->conn_mutex);
    r->conn_cv.notify_one();

    if (status != REDIS_OK) {
        r->state = ERROR;
        log_info("{}, reconnect, after connection error: {}", r->name, ctx->errstr);
        r->reconnect();
    } else {
        r->state = CONNECTED;
        log_info("{}, connected", r->name);
    }
}

void RedisAsync::disconnected(const redisAsyncContext *ctx, int status) {
    auto r = static_cast<RedisAsync*>(ctx->ev.data);

    lock_guard<recursive_mutex> lg(r->conn_mutex);
    r->conn_cv.notify_one();

    /* The context object is always freed by hiredis after the disconnect callback fired.
     * So here we set our `ctx` pointer to null. */
    r->ctx = nullptr;

    if (status != REDIS_OK) {  // disconnection was caused by an error
        r->state = ERROR;
        log_info("{}, reconnect, after disconnection error: {}", r->name, ctx->errstr);
        r->reconnect();
    } else {
        r->state = DISCONNECTED;
        log_info("{}, disconnected", r->name);
    }
}

bool RedisAsync::command(string cmd) {
    lock_guard<recursive_mutex> lg(rw_mutex);
    if (state != CONNECTED)
        return false;
    auto status = redisAsyncCommand(ctx, replied, this, cmd.data());
    return status == REDIS_OK;
}

bool RedisAsync::command(const char *format, ...) {
    lock_guard<recursive_mutex> lg(rw_mutex);
    if (state != CONNECTED)
        return false;

    va_list ap;
    int status;
    va_start(ap,format);
    status = redisvAsyncCommand(ctx, replied, this, format, ap);
    va_end(ap);
    return status == REDIS_OK;
}

void RedisAsync::replied(redisAsyncContext *c, void *r, void *privdata) {
    if (!r) {
        log_warn("redis async command reply null");
        return;
    }
    auto reply = static_cast<redisReply*>(r);
    auto this_ = static_cast<RedisAsync*>(privdata);
    this_->rcallback(reply);
}

void RedisAsync::add_read(void *data) {
    auto r = static_cast<RedisAsync*>(data);
    if (!r->reading) {
        r->reading = true;
        r->loop->add_event(r->fd, EV_READABLE, r->read_handler);
    }
}

void RedisAsync::del_read(void *data) {
    auto r = static_cast<RedisAsync*>(data);
    if (r->reading) {
        r->reading = false;
        r->loop->rmv_event(r->fd, EV_READABLE);
    }
}

void RedisAsync::add_write(void *data) {
    auto r = static_cast<RedisAsync*>(data);
    if (!r->writing) {
        r->writing = true;
        r->loop->add_event(r->fd, EV_WRITABLE, r->write_handler);
    }
}

void RedisAsync::del_write(void *data) {
    auto r = static_cast<RedisAsync*>(data);
    if (r->writing) {
        r->writing = false;
        r->loop->rmv_event(r->fd, EV_WRITABLE);
    }
}

void RedisAsync::cleanup(void *data) {
    auto r = static_cast<RedisAsync*>(data);
    del_read(data);
    del_write(data);
}

Json Event::wait(int timeout) {
    unique_lock<mutex> lk(m);
    bool success = true;
    if (timeout < 0)
        cv.wait(lk);
    else
        success = cv.wait_for(lk, chrono::milliseconds(timeout)) != cv_status::timeout;

    if (!success)
        throw TimeoutError("wait timeout " + std::to_string(timeout));

    return std::move(msg);
}

void Event::notify(Json&& msg) {
    unique_lock<mutex> lk(m);
    this->msg = std::move(msg);
    cv.notify_one();
}

bool RedisSender::publish(string &&channel, Json &&msg) {
    string s = msg.dump();
    if (!command("PUBLISH %b %b", channel.data(), channel.size(), s.data(), s.size()))
        return false;
    if (!channels.count(channel))
        channels.insert(std::move(channel));
    return true;
}

bool RedisSender::push(string &&list, Json &&msg) {
    string s = msg.dump();
    if (!command("LPUSH %b %b", list.data(), list.size(), s.data(), s.size()))
        return false;
    if (!lists.count(list))
        lists.insert(std::move(list));
    return true;
}

void RedisSender::rcallback(redisReply *r) {
    switch (r->type) {
        case REDIS_REPLY_INTEGER:
            assert(r->integer);
            break;
        case REDIS_REPLY_NIL:
            break;
        case REDIS_REPLY_STATUS:
        case REDIS_REPLY_ERROR:
            log_error("RedisSender send error: " + to_str(r));
            break;
        default:
            throw_with_log("RedisSender send reply error type: {}", r->type);
    }
}

void RedisSubscriber::preset(vector<string> &&channels, MsgHandler handler) {
    this->handler = handler;

    cmd_str = "SUBSCRIBE";
    for (auto& c: channels) {
        if (this->channels.count(c)) {
            log_warn("RedisSubscriber subscribe duplicate channel '{}'", c);
            continue;
        }
        this->channels.insert(c);
        cmd_str += " " + c;
    }
    log_info("RedisSubscriber, {}", cmd_str);
}

void RedisSubscriber::rcallback(redisReply *r) {
    switch (r->type) {
        case REDIS_REPLY_ARRAY: {
            assert(r->elements > 0);
            assert(r->element[0]->type == REDIS_REPLY_STRING);
            size_t i = 0;
            while (i < r->elements) {
                assert(i + 2 < r->elements);
                string type = to_str(r->element[i]);
                ++i;
                string channel = to_str(r->element[i]);
                ++i;
                assert(channels.count(channel));
                if (type == "message") {
                    string data = to_str(r->element[i]);

                    if (handler) {
                        try {
                            handler(channel, data);
                        } catch (const exception& e) {
                            log_error(e.what());
                        }
                    }
                }
                else if (type == "subscribe" || type == "unsubscribe") {
                    assert(r->element[i]->type == REDIS_REPLY_INTEGER);
                }
                ++i;
            }
        }
            break;
        case REDIS_REPLY_NIL:
            break;
        case REDIS_REPLY_STATUS:
        case REDIS_REPLY_ERROR:
            log_error("RedisSubscriber subscribe error: " + to_str(r));
            break;
        default:
            throw_with_log("RedisSubscriber subscribe reply error type: {}", r->type);
    }
}

void RedisPoper::preset(unordered_set<string> &&lists, int timeout, MsgHandler handler) {
    this->lists = std::move(lists);

    cmd_str = "BRPOP ";

    for (auto& list: this->lists)
        cmd_str += list + " ";

    /* BRPOP timeout 0 means blocking forever, so here override 0 with 1. */
    if (timeout == 0)
        timeout = 1;

    cmd_str += (timeout < 0 ? "0" : to_string(timeout));
    log_info("RedisPoper, cmd_str '{}'", cmd_str);

    this->handler = handler;
}

void RedisPoper::rcallback(redisReply *r) {
    switch (r->type) {
        case REDIS_REPLY_ARRAY: {
            size_t i = 0;
            while (i < r->elements) {
                assert(i + 1 < r->elements);

                assert(r->element[i]->type == REDIS_REPLY_STRING);
                string list = to_str(r->element[i]);

                ++i;

                assert(r->element[i]->type == REDIS_REPLY_STRING);
                string data = to_str(r->element[i]);

                ++i;

                if (handler) {
                    try {
                        handler(list, data);
                    } catch (const exception& e) {
                        log_error(e.what());
                    }
                }
            }
        }
            break;
        case REDIS_REPLY_NIL:
            if (handler) {
                handler("", {});
            }
            break;
        case REDIS_REPLY_STATUS:
        case REDIS_REPLY_ERROR:
            log_error("RedisPoper pop error: " + to_str(r));
            break;
        default:
            throw_with_log("RedisPoper pop reply error type: {}", r->type);
    }
}

RPCListener::RPCListener(TargetPtr target, const string &ip, int port, EventLoopPtr loop)
    : target(target),
      poper("RPCListener poper", ip, port, loop),
      subscriber("RPCListener subscriber", ip, port, loop),
      sender("RPCListener sender", ip, port, loop)
{
    /* Subscribe for fanout message. */
    subscriber.preset(vector<string>{target->topic},
                      [this](const string& channel, const string& msg) {
                          queue.put(Json::parse(msg), true);
                      });

    /* Pop for call and cast message. */
    unordered_set<string> lists;

    lists.insert(target->topic);  // topic

    if (!target->server.empty())
        lists.insert(target->topic + "." + target->server);  // topic.server

    poper.preset(std::move(lists), POLL_TIMEOUT / 1000,
                 [this](const string& list, const string& msg) {
                     if (!list.empty()) {
                         log_info("poper: {}", msg);
                         try {
                             queue.put(Json::parse(msg), true);
                         } catch (const exception& e) {
                             log_error(e.what());
                         }
                     }

                     if (!stopped)
                         poper.pop();
                 }
    );
}

IncomingMsgPtr RPCListener::poll() {
    auto msg = queue.get(true, 1);
    if (!msg.is_object())
        return nullptr;

    string id = msg["id"];
    string reply_id;
    if (msg.count("reply_id"))
        reply_id = msg["reply_id"].get<string>();

    auto handler = [this, id, reply_id](Json&& reply) mutable {
        log_warn("reply: {}", reply.dump());
        if (!reply_id.empty()) {  // need reply
            Json reply_msg = {
                {"id",  id},
                {"msg", std::move(reply)}
            };
            sender.push(std::move(reply_id), std::move(reply_msg));  // send response
        }
    };

    return IncomingMsgPtr(new IncomingMsg(std::move(msg["msg"]), handler));
}

NotificationListener::
NotificationListener(const vector<TargetPriority> &targets_and_priorities,
                     const string &ip, int port, EventLoopPtr loop)
    : target(target),
      subscriber("NotificationListener subscriber", ip, port, loop)
{
    vector<string> channels;
    for (auto& item: targets_and_priorities) {
        auto &target = item.first;
        auto &priority = item.second;
        channels.push_back(target->topic + "." + priority);  // topic.priority
    }
    subscriber.preset(std::move(channels),
                      [this](const string& channel, const string& msg) {
                          queue.put(Json::parse(msg), true);
                      });
}

IncomingMsgPtr NotificationListener::poll() {
    auto msg = queue.get(true, 1);
    if (!msg.is_object())
        return nullptr;
    return IncomingMsgPtr(new IncomingMsg(std::move(msg)));
}

Sender::Sender(const string &ip, int port, EventLoopPtr loop)
    : sender("Sender sender", ip, port, loop),
      poper("Sender poper", ip, port, loop),
      reply_id(uuid::to_str())
{
    poper.preset(unordered_set<string>({reply_id}), POLL_TIMEOUT / 1000,
                 [this](const string& list, const string& reply) {
                     if (!list.empty()) {
                         try {
                             Json r = Json::parse(reply);
                             string id = r["id"];
                             Json msg = std::move(r["msg"]);
                             {
                                 lock_guard<mutex> ul(events_mutex);
                                 if (events.count(id))
                                     events[id]->notify(std::move(msg));
                             }
                         } catch (const exception& e) {
                             log_error(e.what());
                             return;
                         }
                     }
                     if (!stopped)
                         poper.pop();
                 }
    );
}

void Sender::start() {
    stopped = false;

    sender.start();
    poper.start();

    // Start polling for replies.
    poper.pop();
}

void Sender::stop() {
    stopped = true;

    sender.stop();
    poper.stop();
    events.clear();
}

Json Sender::send(TargetPtr target, const Json &msg,
                  bool wait_for_reply, int timeout, int retry)
{
    // Fanout message can not be waited for reply, and should not have destination server.
    assert(!(wait_for_reply && target->fanout));
    assert(!(!target->server.empty() && target->fanout));

    // Generate uuid as the message id.
    string id = uuid::to_str();

    Json data = {
        {"id",  id},
        {"msg", msg}
    };

    if (wait_for_reply)
        data["reply_id"] = reply_id;  // need reply

    if (target->fanout)
        publish(target->topic, std::move(data));
    else {
        string topic = target->topic + (target->server.empty() ? "" : "." + target->server);
        if (!push(topic, data))
            log_error("redis command fail");
        // Waiting for reply will block until timeout.
        if (wait_for_reply) {
            auto event = make_shared<Event>();

            {
                lock_guard<mutex> ul(events_mutex);
                events[id] = event;
            }

            auto events_guard = make_guard([this, &id] {
                lock_guard<mutex> ul(events_mutex);
                events.erase(id);
            });

            Json reply;
            while (true) {
                try {
                    reply = event->wait(timeout * 1000);
                    break;  // received reply successfully
                } catch (const TimeoutError &e) {
                    assert(timeout >= 0);
                    if (retry > 0) {
                        --retry;
                        push(topic, data);
                    }
                    else if (retry == 0)
                        // really timeout after retries
                        throw TimeoutError(fmt::format("RPC send {}, msg id {}", e.what(), id));
                    log_warn("retry sending messaging and waiting for reply");
                }
            }

            if (reply.count("error"))
                throw CallError(reply["error"].dump());
            return reply["reply"];
        }
    }

    return Json();
}

void RedisDriver::start() {
    log_info("=============== messaging redis driver start ===============");
    loop->start();
    sender.start();
    log_info("=============== messaging redis driver started ===============");
}

void RedisDriver::stop() {
    log_info("=============== messaging redis driver stop ===============");
    sender.stop();
    for (auto& listener: listeners)
        listener->stop();
    loop->stop();
    log_info("=============== messaging redis driver stopped ===============");
}

}}
